package com.pai4j.act.service;


import com.fasterxml.jackson.core.type.TypeReference;
import com.pai4j.act.repository.dao.ICommentDAO;
import com.pai4j.act.repository.dao.IResourceBehaviorDAO;
import com.pai4j.act.repository.entity.CommentEntity;
import com.pai4j.act.repository.entity.ResourceBehaviorEntity;
import com.pai4j.common.bean.PAIPageResponseBeanUtil;
import com.pai4j.common.bean.PAIResponseBean;
import com.pai4j.common.bean.PAIResponseBeanUtil;
import com.pai4j.common.constants.ResourceBehaviorConstants;
import com.pai4j.common.enums.MessageTypeEnum;
import com.pai4j.common.enums.CommonStatusEnum;
import com.pai4j.common.enums.ResponseCodeEnum;
import com.pai4j.common.enums.messagequeue.MessageBroadChannelEnum;
import com.pai4j.common.helper.SessionHelper;
import com.pai4j.common.service.messagequeue.producer.MessageQueueProducer;
import com.pai4j.common.util.*;
import com.pai4j.domain.vo.request.mess.MessageNotifyVO;
import com.pai4j.domain.vo.request.request.CommentSaveVO;
import com.pai4j.domain.vo.response.CommentOutResponseVO;
import com.pai4j.domain.vo.response.CommentResponseVO;
import com.pai4j.domain.vo.response.IpLocationResponseVO;
import com.pai4j.domain.vo.response.UserBaseResponseInfoVO;
import com.pai4j.domain.vo.response.moment.MomentVO;
import com.pai4j.domain.vo.response.subjectaccount.ArticleBaseResponseVO;
import com.pai4j.remote.pgc.SubjectArticleClient;
import com.pai4j.remote.ugc.MomentClient;
import jakarta.persistence.criteria.CriteriaBuilder;
import jakarta.persistence.criteria.CriteriaQuery;
import jakarta.persistence.criteria.Predicate;
import jakarta.persistence.criteria.Root;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.data.jpa.domain.Specification;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import com.pai4j.common.enums.RedisKeyEnum;
import java.util.*;
import java.util.stream.Collectors;


@Slf4j
@Service
public class CommentService {
    
    
    @Autowired
    private ICommentDAO commentDAO;
    @Autowired
    private IResourceBehaviorDAO resourceBehaviorDAO;
    @Autowired
    private AccountService userService;
    @Autowired
    private RedisUtil redisUtil;
    @Autowired
    private MessageQueueProducer<MessageNotifyVO, Long> messageQueueProducer;
    @Autowired
    private MomentClient momentClient;
    @Autowired
    private SubjectArticleClient subjectArticleClient;
    
    /**
     * 默认回复评论条数
     */
    private static final int DEFAULT_REPLY_COMMENT_SIZE = 20;
    
    /**
     * 默认外显评论条数
     */
    private static final int DEFAULT_OUT_COMMENT_SIZE = 3;
    
    /***
     * 发布评论
     */
    public Long saveComment(CommentSaveVO commentSaveVO) {
        Long replyId = commentSaveVO.getReplyId();
        Long parentId = null;
        if (replyId != null) {
            CommentEntity replyComment = commentDAO.findAllById(replyId);
            Assert.isTrue(replyComment != null, "回复评论不存在!");
            if (replyComment.getParentId() != null) {
                parentId = replyComment.getParentId();
            } else {
                parentId = replyComment.getId();
            }
        }
        CommentEntity commentEntity = new CommentEntity();
        commentEntity.setResourceId(commentSaveVO.getResourceId());
        commentEntity.setResourceType(commentSaveVO.getResourceType());
        commentEntity.setAuthor(commentSaveVO.getAuthorId());
        String commentContent = commentSaveVO.getContent();
        commentEntity.setContent(commentContent);
        commentEntity.setImages(commentSaveVO.getImages());
        commentEntity.setPubDate(new Date());
        commentEntity.setReplyId(replyId);
        commentEntity.setParentId(parentId);
        commentEntity.setIsTop(false);
        commentEntity.setLikeCount(0L);
        commentEntity.setStatus(CommonStatusEnum.PUBLISHED.getStatus());
        
        // 获取客户端IP
        String ip = commentSaveVO.getIp();
        String ipAddress = null;
        // 成功获取客户端IP，解析IP归属地
        if (StringUtils.isNotBlank(ip)) {
            IpLocationResponseVO ipAddressVO = IPAddressUtil.location(ip);
            if (ipAddressVO != null && StringUtils.isNotBlank(ipAddressVO.getProvince())) {
                // IP归属地取IP地址位置具体到省份
                ipAddress = StringUtils.isBlank(ipAddressVO.getProvince()) ? "未知" : ipAddressVO.getProvince();
            }
        }
        commentEntity.setIp(ip);
        commentEntity.setIpAddress(ipAddress);
        Long commentId = commentDAO.save(commentEntity).getId();
        //回复id为空，说明是首次评论，需要刷新评论数量缓存
        if (replyId == null) {
            // 刷新评论数量
            addCommentCount2Cache(commentSaveVO.getResourceType(), commentSaveVO.getResourceId());
            // 发送评论消息 : 用户A评论了资源B，用户B收到消息
            sendCommentNotification(commentSaveVO.getAuthorId(), commentSaveVO.getResourceType(), commentSaveVO.getResourceId(), commentSaveVO.getContent(),commentId);
        } else {
            // 发送评论回复消息: 用户A回复了用户B的评论，用户B收到消息
            // id是回复的评论本身的自增id
            CommentEntity replyComment = commentDAO.findAllById(commentId);
            if (replyComment != null) {
                String authorId = commentDAO.findAllById(replyComment.getReplyId()).getAuthor();
                sendReplyNotification(authorId, replyComment);
            }
        }
        // 刷新外显评论缓存
        ThreadPoolExecutorUtil.execute(() ->
                refreshOutCommentCache(commentSaveVO.getResourceType(), commentSaveVO.getResourceId()));
        return commentId;
    }
    
    
    /**
     *   删除评论
     *
     * @author: CYM-pai
     * @date: 2025/07/30 18:18
     **/
    public void deleteComment(Long commentId, String userId) {
        CommentEntity commentEntity = commentDAO.findAllById(commentId);
        commentEntity.setStatus(CommonStatusEnum.DELETED.getStatus());
        commentEntity.setUpdateBy(userId);
        commentDAO.save(commentEntity);
        if (commentEntity.getReplyId() == null) {
            // 刷新评论数量
            reduceCommentCount2Cache(commentEntity.getResourceType(), commentEntity.getResourceId());
        }
        log.info("{} 删除了评论 {}!", userId, commentId);
        // 刷新外显评论缓存
        ThreadPoolExecutorUtil.execute(() ->
                refreshOutCommentCache(commentEntity.getResourceType(), commentEntity.getResourceId()));
    }
    
    /***
     * 批量查询外显评论
     * @param resourceType
     * @param resourceIdList
     * @return
     */
    public Map<Long, List<CommentOutResponseVO>> batchGetOutCommentFromCache(String resourceType, List<Long> resourceIdList) {
        Map<Long, List<CommentOutResponseVO>> bulkQueryResultMap = new HashMap<>();
        if (CollectionUtils.isEmpty(resourceIdList)) {
            return bulkQueryResultMap;
        }
        List<String> bulkKey = resourceIdList.stream()
                .map(rid -> RedisKeyEnum.COMMENT_OUT_LIST_CACHE.getKey(resourceType, String.valueOf(rid)))
                .collect(Collectors.toList());
        List<String> bulkQueryResult = redisUtil.mget(bulkKey);
        for (int i = 0; i < resourceIdList.size(); i++) {
            String val = bulkQueryResult.get(i);
            Long resourceId = resourceIdList.get(i);
            List<CommentOutResponseVO> commentOutResponseVOList;
            if (StringUtils.isBlank(val)) {
                commentOutResponseVOList = refreshOutCommentCache(resourceType, resourceId);
            } else {
                commentOutResponseVOList = JsonUtil.fromJson(val, new TypeReference<List<CommentOutResponseVO>>() {
                });
            }
            for (CommentOutResponseVO commentOutResponseVO : commentOutResponseVOList) {
                String author = commentOutResponseVO.getAuthor();
                String replyAuthor = commentOutResponseVO.getReplyAuthor();
                if (StringUtils.isNotBlank(author)) {
                    commentOutResponseVO.setAuthorInfo(userService.accountInfo(author));
                }
                if (StringUtils.isNotBlank(replyAuthor)) {
                    commentOutResponseVO.setReplyAuthorInfo(userService.accountInfo(replyAuthor));
                }
            }
            bulkQueryResultMap.put(resourceId, commentOutResponseVOList);
        }
        return bulkQueryResultMap;
    }
    
    /***
     * 查询外显评论
     * @param resourceType
     * @param resourceId
     * @return
     */
    public List<CommentOutResponseVO> getOutCommentFromCache(String resourceType, Long resourceId) {
        String key = RedisKeyEnum.COMMENT_OUT_LIST_CACHE.getKey(resourceType, String.valueOf(resourceId));
        String val = redisUtil.get(key);
        List<CommentOutResponseVO> commentOutResponseVOList;
        if (StringUtils.isBlank(val)) {
            commentOutResponseVOList = refreshOutCommentCache(resourceType, resourceId);
        } else {
            commentOutResponseVOList = JsonUtil.fromJson(val, new TypeReference<List<CommentOutResponseVO>>() {
            });
        }
        for (CommentOutResponseVO commentOutResponseVO : commentOutResponseVOList) {
            String author = commentOutResponseVO.getAuthor();
            String replyAuthor = commentOutResponseVO.getReplyAuthor();
            if (StringUtils.isNotBlank(author)) {
                commentOutResponseVO.setAuthorInfo(userService.accountInfo(author));
            }
            if (StringUtils.isNotBlank(replyAuthor)) {
                commentOutResponseVO.setReplyAuthorInfo(userService.accountInfo(replyAuthor));
            }
        }
        return commentOutResponseVOList;
    }
    
    /***
     * 刷新外显评论
     * @param resourceType
     * @param resourceId
     * @return
     */
    private List<CommentOutResponseVO> refreshOutCommentCache(String resourceType, Long resourceId) {
        if (!ResourceBehaviorConstants.ResourceType.MOMENT.name().equals(resourceType)) {
            return Collections.emptyList();
        }
        String key = RedisKeyEnum.COMMENT_OUT_LIST_CACHE.getKey(resourceType, String.valueOf(resourceId));
        List<CommentEntity> commentEntities = commentDAO.getOutSideCommentListLimit(resourceType, resourceId,
                CommonStatusEnum.PUBLISHED.getStatus(), DEFAULT_OUT_COMMENT_SIZE);
        List<CommentOutResponseVO> commentOutResponseVOList = new ArrayList<>();
        if (org.apache.commons.collections.CollectionUtils.isNotEmpty(commentEntities)) {
            commentEntities.forEach(c -> {
                CommentOutResponseVO commentOutResponseVO = new CommentOutResponseVO();
                commentOutResponseVO.setId(c.getId());
                String content = c.getContent() == null ? "" : c.getContent();
                if (StringUtils.isNotBlank(c.getImages())) {
                    for (String image : c.getImages().split(",")) {
                        content += " [图片]";
                    }
                }
                commentOutResponseVO.setContent(content);
                commentOutResponseVO.setAuthor(c.getAuthor());
                commentOutResponseVO.setIpAddress(c.getIpAddress());
                if (c.getReplyId() != null) {
                    CommentEntity replyComment = commentDAO.findAllById(c.getReplyId());
                    commentOutResponseVO.setReplyAuthor(replyComment.getAuthor());
                }
                commentOutResponseVOList.add(commentOutResponseVO);
            });
        }
        redisUtil.set(key, JsonUtil.toJsonString(commentOutResponseVOList), RedisKeyEnum.COMMENT_OUT_LIST_CACHE.getExpireTime());
        return commentOutResponseVOList;
    }
    
    /***
     * 刷新评论数量
     * @param resourceType
     * @param resourceId
     */
    public Long getCommentCountFromCache(Long resourceId, String resourceType) {
        String resourceIdStr = String.valueOf(resourceId);
        String key = RedisKeyEnum.COMMENT_COUNT_CACHE.getKey(resourceType);
        String countCache = redisUtil.hget(key, resourceIdStr);
        if (StringUtils.isNotBlank(countCache)) {
            return Long.valueOf(countCache);
        }
        return refreshResourceCommentCountCache(resourceId, resourceType);
    }
    
    public Map<Long, Long> batchGetCommentCountFromCache(List<Long> resourceIdList, String resourceType) {
        Map<Long, Long> resultMap = new HashMap<>();
        String key = RedisKeyEnum.COMMENT_COUNT_CACHE.getKey(resourceType);
        List<String> resourceIdKeys = resourceIdList.stream().map(String::valueOf).collect(Collectors.toList());
        List<String> resourceVals = redisUtil.hmget(key, resourceIdKeys);
        for (int i = 0; i < resourceIdList.size(); i++) {
            Long id = resourceIdList.get(i);
            String cache = resourceVals.get(i);
            Long count = 0L;
            if (StringUtils.isNotBlank(cache)) {
                count = Long.valueOf(cache);
            } else {
                count = refreshResourceCommentCountCache(id, resourceType);
            }
            resultMap.put(id, count);
        }
        return resultMap;
    }
    
    /***
     * 刷新评论数量
     * @param resourceType
     * @param resourceId
     */
    private void addCommentCount2Cache(String resourceType, Long resourceId) {
        String resourceIdStr = String.valueOf(resourceId);
        String key = RedisKeyEnum.COMMENT_COUNT_CACHE.getKey(resourceType);
        Long count = redisUtil.hincrex(key, resourceIdStr);
        if (count == 1) {
            refreshResourceCommentCountCache(resourceId, resourceType);
        }
    }
    
    /***
     * 刷新评论数量
     * @param resourceType
     * @param resourceId
     */
    private void reduceCommentCount2Cache(String resourceType, Long resourceId) {
        String resourceIdStr = String.valueOf(resourceId);
        String key = RedisKeyEnum.COMMENT_COUNT_CACHE.getKey(resourceType);
        Long count = redisUtil.hdecrex(key, resourceIdStr);
        if (count == 0) {
            refreshResourceCommentCountCache(resourceId, resourceType);
        }
    }
    
    private Long refreshResourceCommentCountCache(Long resourceId, String resourceType) {
        String key = RedisKeyEnum.COMMENT_COUNT_CACHE.getKey(resourceType);
        Long count = getResourceCommentCountFromDB(resourceId, resourceType);
        redisUtil.hset(key, String.valueOf(resourceId), String.valueOf(count), RedisKeyEnum.COMMENT_COUNT_CACHE.getExpireTime());
        return count;
    }
    
    private Long getResourceCommentCountFromDB(Long resourceId, String resourceType) {
        return commentDAO.countByResourceIdAndResourceTypeAndStatusAndParentIdIsNull(resourceId, resourceType, CommonStatusEnum.PUBLISHED.getStatus());
    }
    
    /***
     * 评论置顶
     * @param commentId
     */
    public void topComment(Long commentId) {
        CommentEntity commentEntity = commentDAO.findAllById(commentId);
        Assert.isTrue(commentEntity != null, "评论不存在");
        commentEntity.setIsTop(true);
        commentEntity.setTopDate(new Date());
        commentDAO.save(commentEntity);

        // 发送置顶通知消息
        String currentUserId = SessionHelper.getCurrentUserId();
        sendTopCommentNotification(currentUserId, commentEntity);

        // 刷新外显评论缓存
        ThreadPoolExecutorUtil.execute(() ->
                refreshOutCommentCache(commentEntity.getResourceType(), commentEntity.getResourceId()));
    }
    
    /***
     * 取消评论置顶
     * @param commentId
     */
    public void cancelTopComment(Long commentId) {
        CommentEntity commentEntity = commentDAO.findAllById(commentId);
        Assert.isTrue(commentEntity != null, "评论不存在");
        commentEntity.setIsTop(false);
        commentEntity.setTopDate(null);
        commentDAO.save(commentEntity);

        // 刷新外显评论缓存
        ThreadPoolExecutorUtil.execute(() ->
                refreshOutCommentCache(commentEntity.getResourceType(), commentEntity.getResourceId()));
    }
    
    /***
     * 按条件查询评论列表
     * @return
     */
    public PAIPageResponseBeanUtil<List<CommentResponseVO>> queryCommentByCondition(
            String content, Long resourceId, String resourceType, Integer pageNo, Integer pageSize, boolean isHot,
            boolean isNew) {
        // 排序规则
        Sort sort = buildSortCondition(isHot, isNew);
        // 分页条件
        Pageable pageable = PageRequest.of(pageNo - 1, pageSize, sort);
        // 构建查询条件
        Specification<CommentEntity> specification = buildCommentQuerySpecification(content, resourceId, resourceType);
        // 查询评论
        Page<CommentEntity> commentEntityPage = commentDAO.findAll(specification, pageable);
        
        List<CommentResponseVO> commentResponseVOList = new ArrayList<>();
        if (commentEntityPage != null && commentEntityPage.getTotalElements() > 0) {
            for (CommentEntity commentEntity : commentEntityPage.getContent()) {
                CommentResponseVO vo = convertCommentEntityToVo(commentEntity);
                vo.setReplyToList(getCommentReplyCommentVoList(vo.getId()));
                commentResponseVOList.add(vo);
            }
        }
        return PAIPageResponseBeanUtil.success(pageNo, pageSize, commentEntityPage.getTotalElements(), commentResponseVOList);
    }
    
    private Sort buildSortCondition(boolean isHot, boolean isNew) {
        List<Sort.Order> orderList = new ArrayList<>();
        orderList.add(new Sort.Order(Sort.Direction.DESC, "topDate"));
        if (isNew) {
            orderList.add(new Sort.Order(Sort.Direction.DESC, "id"));
        }
        if (isHot) {
            orderList.add(new Sort.Order(Sort.Direction.DESC, "likeCount"));
        }
        return Sort.by(orderList);
    }
    
    private Specification<CommentEntity> buildCommentQuerySpecification(String content, Long resourceId,
                                                                        String resourceType) {
        
        Specification<CommentEntity> specification = new Specification<CommentEntity>() {
            @Override
            public Predicate toPredicate(Root<CommentEntity> root, CriteriaQuery<?> criteriaQuery, CriteriaBuilder criteriaBuilder) {
                List<Predicate> predicates = new ArrayList<>();
                predicates.add(criteriaBuilder.equal(root.get("status").as(String.class), "PUBLISHED"));
                predicates.add(criteriaBuilder.isNull(root.get("parentId").as(Long.class)));
                if (resourceId != null) {
                    predicates.add(criteriaBuilder.equal(root.get("resourceId").as(Long.class), resourceId));
                }
                if (StringUtils.isNotBlank(resourceType)) {
                    predicates.add(criteriaBuilder.equal(root.get("resourceType").as(String.class), resourceType));
                }
                if (StringUtils.isNotBlank(content)) {
                    predicates.add(
                            criteriaBuilder.or(
                                    criteriaBuilder.like(root.get("content").as(String.class), "%" + content + "%")));
                }
                Predicate[] pre = new Predicate[predicates.size()];
                criteriaQuery.where(predicates.toArray(pre));
                return criteriaQuery.getRestriction();
            }
        };
        return specification;
    }
    
    private List<CommentResponseVO> getCommentReplyCommentVoList(Long parentId) {
        List<CommentResponseVO> commentResponseVOList = new ArrayList<>();
        List<CommentEntity> replyComments =
                commentDAO.findAllByParentIdAndStatus(parentId, CommonStatusEnum.PUBLISHED.getStatus(), DEFAULT_REPLY_COMMENT_SIZE);
        if (CollectionUtils.isEmpty(replyComments)) {
            return commentResponseVOList;
        }
        for (CommentEntity commentEntity : replyComments) {
            commentResponseVOList.add(convertCommentEntityToVo(commentEntity));
        }
        return commentResponseVOList;
    }
    
    private CommentResponseVO convertCommentEntityToVo(CommentEntity commentEntity) {
        CommentResponseVO commentResponseVO = new CommentResponseVO();
        BeanUtils.copyProperties(commentEntity, commentResponseVO);
        commentResponseVO.setPubDateStr(DateUtils.getDate2String(commentEntity.getPubDate()));
        commentResponseVO.setAuthorInfo(userService.accountInfo(commentEntity.getAuthor()));
        if (commentEntity.getReplyId() != null) {
            CommentEntity replyEntity = commentDAO.findAllById(commentEntity.getReplyId());
            if (replyEntity != null) {
                CommentResponseVO replyVO = new CommentResponseVO();
                BeanUtils.copyProperties(replyEntity, replyVO);
                replyVO.setPubDateStr(DateUtils.getDate2String(replyEntity.getPubDate()));
                replyVO.setAuthorInfo(userService.accountInfo(replyEntity.getAuthor()));
                commentResponseVO.setReply(replyVO);
            }
        }
        return commentResponseVO;
    }
    
    /**
     * 发送评论通知消息
     *
     * @param userId       评论者用户ID
     * @param resourceType 资源类型
     * @param resourceId   资源ID
     */
    private void  sendCommentNotification(String userId, String resourceType, Long resourceId, String commentContent, Long commentId) {
        try {
            // 获取资源拥有者
            String resourceOwner = getResourceOwner(resourceType, resourceId);
            if (StringUtils.isBlank(resourceOwner)) {
                return;
            }
            
            // 不给自己发送通知
            if (ObjectUtils.equals(userId, resourceOwner)) {
                return;
            }
            
            // 获取评论者信息
            UserBaseResponseInfoVO senderInfo = userService.accountInfo(userId);
            if (senderInfo == null) {
                return;
            }
            
            // 构建通知消息
            MessageNotifyVO messageNotifyVO = new MessageNotifyVO();
            messageNotifyVO.setSenderId(userId);
            messageNotifyVO.setReceiverId(resourceOwner);
            messageNotifyVO.setSender(senderInfo);
            messageNotifyVO.setTime(DateUtils.getCurrentTimeMillis());
            messageNotifyVO.setType(MessageTypeEnum.INTERACTION.getType());
            
            // 设置互动消息类型为评论
            messageNotifyVO.setMessageType(com.pai4j.common.enums.InteractionMessageTypeEnum.COMMENT.getCode());
            
            // 设置评论内容和ID
            messageNotifyVO.setContent(commentContent);
            messageNotifyVO.setCommentId(commentId);
            
            // 设置资源信息
            messageNotifyVO.setResourceId(resourceId);
            messageNotifyVO.setResourceType(resourceType);
            
            // 填充资源详细信息（标题、封面图、跳转链接）
            fillResourceDetails(messageNotifyVO, resourceType, resourceId);
            
            // 构建消息内容
            String message = buildNotificationMessage(senderInfo.getUserName(), "评论", resourceType);
            messageNotifyVO.setMessage(message);
            
            // 发送通知
            messageQueueProducer.broadSend(MessageBroadChannelEnum.QUEUE_INTERACTION_NOTIFY, messageNotifyVO);
            
        } catch (Exception e) {
            // 通知发送失败不影响主业务，只记录日志
            log.error("发送评论通知失败: userId={}, resourceType={}, resourceId={}", userId, resourceType, resourceId, e);
        }
    }
    
    /**
     * 发送回复通知消息
     *
     * @param userId       被回复者用户ID
     * @param replyComment 被回复的评论
     */
    private void sendReplyNotification(String userId, CommentEntity replyComment) {
        try {
            String replyAuthor = replyComment.getAuthor();
            if (StringUtils.isBlank(replyAuthor)) {
                return;
            }
            
            // 不给自己发送通知
            if (ObjectUtils.equals(userId, replyAuthor)) {
                return;
            }
            
            // 获取回复者信息
            UserBaseResponseInfoVO senderInfo = userService.accountInfo(replyAuthor);
            if (senderInfo == null) {
                return;
            }
            Long commentId = replyComment.getReplyId();
            String commentContent = commentDAO.findAllById(commentId).getContent();
            // 构建通知消息
            MessageNotifyVO messageNotifyVO = new MessageNotifyVO();
            messageNotifyVO.setSenderId(replyAuthor);
            messageNotifyVO.setReceiverId(userId);
            messageNotifyVO.setSender(senderInfo);
            messageNotifyVO.setTime(DateUtils.getCurrentTimeMillis());
            messageNotifyVO.setType(MessageTypeEnum.INTERACTION.getType());
            
            // 设置互动消息类型为回复
            messageNotifyVO.setMessageType(com.pai4j.common.enums.InteractionMessageTypeEnum.REPLY.getCode());
            
            // 设置回复内容和ID
            messageNotifyVO.setContent(replyComment.getContent());
            messageNotifyVO.setMessageExt(commentContent);
            messageNotifyVO.setReplyId(replyComment.getReplyId());
            messageNotifyVO.setCommentId(commentId);  // 同时设置commentId
            
            // 设置资源信息（从CommentEntity中获取）
            messageNotifyVO.setResourceId(replyComment.getResourceId());
            messageNotifyVO.setResourceType(replyComment.getResourceType());
            
            // 填充资源详细信息
            if (replyComment.getResourceId() != null && replyComment.getResourceType() != null) {
                fillResourceDetails(messageNotifyVO, replyComment.getResourceType(), replyComment.getResourceId());
            }
            
            // 构建消息内容
            String message = String.format("%s 回复了你的评论: %s", senderInfo.getUserName(), messageNotifyVO.getMessageExt());
            messageNotifyVO.setMessage(message);
            
            // 发送通知
            messageQueueProducer.broadSend(MessageBroadChannelEnum.QUEUE_INTERACTION_NOTIFY, messageNotifyVO);
            
        } catch (Exception e) {
            // 通知发送失败不影响主业务，只记录日志
            log.error("发送回复通知失败: userId={}, replyCommentId={}", userId, replyComment.getId(), e);
        }
    }
    
    /**
     * 根据资源类型获取资源拥有者
     *
     * @param resourceType 资源类型
     * @param resourceId   资源ID
     * @return 资源拥有者用户ID
     */
    private String getResourceOwner(String resourceType, Long resourceId) {
        try {
            switch (resourceType) {
                case "MOMENT":
                    PAIResponseBean<MomentVO> momentResponse = momentClient.detail(resourceId);
                    if (PAIResponseBeanUtil.isOk(momentResponse) && momentResponse.getData() != null) {
                        return momentResponse.getData().getAuthor();
                    }
                    break;
                case "ARTICLE":
                    PAIResponseBean<ArticleBaseResponseVO> articleResponse = subjectArticleClient.detail(resourceId, false);
                    if (PAIResponseBeanUtil.isOk(articleResponse) && articleResponse.getData() != null) {
                        return articleResponse.getData().getAuthor();
                    }
                    break;
                default:
                    // 其他资源类型可以在这里扩展
                    break;
            }
        } catch (Exception e) {
            log.error("获取资源拥有者失败: resourceType={}, resourceId={}", resourceType, resourceId, e);
        }
        return null;
    }
    
    /**
     * 构建通知消息内容
     *
     * @param senderName   发送者姓名
     * @param actionType   操作类型
     * @param resourceType 资源类型
     * @return 通知消息内容
     */
    private String buildNotificationMessage(String senderName, String actionType, String resourceType) {
        String resourceTypeName;
        switch (resourceType) {
            case "MOMENT":
                resourceTypeName = "π圈动态";
                break;
            case "ARTICLE":
                resourceTypeName = "专题文章";
                break;
            case "COURSE":
                resourceTypeName = "课程";
                break;
            default:
                resourceTypeName = "内容";
                break;
        }
        return String.format("%s %s了你的%s", senderName, actionType, resourceTypeName);
    }

    /**
     * 发送置顶评论通知消息
     *
     * @param operatorId     操作者用户ID（置顶操作者）
     * @param commentEntity  被置顶的评论
     */
    private void sendTopCommentNotification(String operatorId, CommentEntity commentEntity) {
        try {
            String commentAuthor = commentEntity.getAuthor();
            if (StringUtils.isBlank(commentAuthor)) {
                return;
            }

            // 不给自己发送通知
            if (ObjectUtils.equals(operatorId, commentAuthor)) {
                return;
            }

            // 获取操作者信息
            UserBaseResponseInfoVO operatorInfo = userService.accountInfo(operatorId);
            if (operatorInfo == null) {
                return;
            }

            // 构建通知消息
            MessageNotifyVO messageNotifyVO = new MessageNotifyVO();
            messageNotifyVO.setSenderId(operatorId);
            messageNotifyVO.setReceiverId(commentAuthor);
            messageNotifyVO.setSender(operatorInfo);
            messageNotifyVO.setTime(DateUtils.getCurrentTimeMillis());
            messageNotifyVO.setType(MessageTypeEnum.INTERACTION.getType());

            // 构建消息内容 - 根据资源类型构建不同的消息
            String resourceTypeName = getResourceTypeName(commentEntity.getResourceType());
            String message = String.format("%s 置顶了你在%s下的评论",
                    operatorInfo.getUserName(), resourceTypeName);
            messageNotifyVO.setMessage(message);

            // 发送通知
            messageQueueProducer.broadSend(MessageBroadChannelEnum.QUEUE_INTERACTION_NOTIFY, messageNotifyVO);

        } catch (Exception e) {
            // 通知发送失败不影响主业务，只记录日志
            log.error("发送置顶评论通知失败: operatorId={}, commentId={}", operatorId, commentEntity.getId(), e);
        }
    }

    /**
     * 获取资源类型名称
     *
     * @param resourceType 资源类型
     * @return 资源类型名称
     */
    private String getResourceTypeName(String resourceType) {
        switch (resourceType) {
            case "MOMENT":
                return "π圈动态";
            case "ARTICLE":
                return "专题文章";
            case "COURSE":
                return "课程";
            default:
                return "内容";
        }
    }

    /**
     * 批量查询评论数量（走缓存）
     * 该方法通过Redis缓存批量获取多个资源的评论数量，提高查询性能
     * 
     * @param resourceIds 资源ID列表
     * @param resourceType 资源类型
     * @return 资源ID与评论数量的映射关系
     */
    public PAIResponseBean<Map<Long, Long>> batchCountFromCache(List<Long> resourceIds, String resourceType) {
        // 参数校验
        if (CollectionUtils.isEmpty(resourceIds)) {
            log.warn("批量查询评论数量: resourceIds为空");
            return PAIResponseBeanUtil.success(new HashMap<>());
        }
        
        if (StringUtils.isBlank(resourceType)) {
            log.warn("批量查询评论数量: resourceType为空");
            return PAIResponseBeanUtil.error(ResponseCodeEnum.ERROR.getCode(), "资源类型不能为空");
        }

        try {
            Map<Long, Long> resultMap = new HashMap<>();
            
            // 构建Redis缓存键列表
            List<String> cacheKeys = resourceIds.stream()
                    .map(resourceId -> buildCommentCountCacheKey(resourceType, resourceId))
                    .collect(Collectors.toList());
            
            // 批量从Redis获取缓存数据
            List<String> cachedCounts = redisUtil.mget(cacheKeys);
            
            // 记录缓存未命中的资源ID
            List<Long> missedResourceIds = new ArrayList<>();
            
            // 处理缓存结果
            for (int i = 0; i < resourceIds.size(); i++) {
                Long resourceId = resourceIds.get(i);
                String cachedCount = cachedCounts.get(i);
                
                if (StringUtils.isNotBlank(cachedCount)) {
                    // 缓存命中，直接使用缓存值
                    try {
                        resultMap.put(resourceId, Long.parseLong(cachedCount));
                    } catch (NumberFormatException e) {
                        log.warn("缓存数据格式错误: resourceId={}, cachedCount={}", resourceId, cachedCount);
                        missedResourceIds.add(resourceId);
                    }
                } else {
                    // 缓存未命中，记录需要从数据库查询的资源ID
                    missedResourceIds.add(resourceId);
                }
            }
            
            // 对于缓存未命中的资源，从数据库查询并更新缓存
            if (!CollectionUtils.isEmpty(missedResourceIds)) {
                for (Long resourceId : missedResourceIds) {
                    Long count = commentDAO.countByResourceIdAndResourceTypeAndStatusAndParentIdIsNull(
                            resourceId, resourceType, CommonStatusEnum.PUBLISHED.getStatus());
                    
                    if (count == null) {
                        count = 0L;
                    }
                    
                    resultMap.put(resourceId, count);
                    
                    // 更新缓存，设置1小时过期时间
                    String cacheKey = buildCommentCountCacheKey(resourceType, resourceId);
                    redisUtil.set(cacheKey, count.toString(), 3600L);
                }
                
                log.info("批量查询评论数量: 缓存命中={}, 数据库查询={}", 
                        resourceIds.size() - missedResourceIds.size(), missedResourceIds.size());
            } else {
                log.info("批量查询评论数量: 全部缓存命中, 查询数量={}", resourceIds.size());
            }
            
            return PAIResponseBeanUtil.success(resultMap);
            
        } catch (Exception e) {
            log.error("批量查询评论数量异常: resourceIds={}, resourceType={}", 
                    JsonUtil.toJsonString(resourceIds), resourceType, e);
            return PAIResponseBeanUtil.error(ResponseCodeEnum.ERROR.getCode(),"批量查询评论数量失败");
        }
    }
    
    /**
     * 构建评论数量缓存键
     * 
     * @param resourceType 资源类型
     * @param resourceId 资源ID
     * @return 缓存键
     */
    private String buildCommentCountCacheKey(String resourceType, Long resourceId) {
        return String.format("comment:count:%s:%d", resourceType, resourceId);
    }

    /**
     * 获取资源点赞用户列表（支持分页）
     * 该方法查询指定资源的点赞用户，并返回用户基本信息（包含头像）
     * 
     * @param resourceId 资源ID
     * @param resourceType 资源类型
     * @param pageNo 页码
     * @param pageSize 每页大小
     * @return 点赞用户信息列表
     */
    public PAIResponseBean<List<UserBaseResponseInfoVO>> getLikeUsersByResource(
            Long resourceId, String resourceType, Integer pageNo, Integer pageSize) {
        
        try {
            // 1. 查询资源的所有点赞行为记录
            List<ResourceBehaviorEntity> likeRecords = resourceBehaviorDAO.findAllByResourceBehaviorUser(
                    resourceId, ResourceBehaviorConstants.BehaviorType.LIKE.name(), resourceType);
            
            if (CollectionUtils.isEmpty(likeRecords)) {
                log.info("资源暂无点赞用户: resourceId={}, resourceType={}", resourceId, resourceType);
                return PAIResponseBeanUtil.success(new ArrayList<>());
            }
            
            // 2. 提取用户ID列表并按时间倒序排序（最新点赞的用户在前）
            List<String> userIds = likeRecords.stream()
                    .sorted((a, b) -> b.getUpdateDate().compareTo(a.getUpdateDate()))
                    .map(ResourceBehaviorEntity::getUserId)
                    .distinct()
                    .collect(Collectors.toList());
            
            // 3. 分页处理
            int startIndex = (pageNo - 1) * pageSize;
            int endIndex = Math.min(startIndex + pageSize, userIds.size());
            
            if (startIndex >= userIds.size()) {
                log.info("分页超出范围: resourceId={}, resourceType={}, pageNo={}, totalUsers={}", 
                        resourceId, resourceType, pageNo, userIds.size());
                return PAIResponseBeanUtil.success(new ArrayList<>());
            }
            
            List<String> pagedUserIds = userIds.subList(startIndex, endIndex);
            
            // 4. 批量获取用户基本信息（包含头像）
            Map<String, UserBaseResponseInfoVO> userInfoMap = userService.batchGet(new HashSet<>(pagedUserIds));
            
            if (userInfoMap == null || userInfoMap.isEmpty()) {
                log.warn("批量获取用户信息失败: userIds={}", JsonUtil.toJsonString(pagedUserIds));
                return PAIResponseBeanUtil.error(ResponseCodeEnum.ERROR.getCode(),"获取用户信息失败");
            }
            
            // 5. 按照点赞时间顺序构建返回结果
            List<UserBaseResponseInfoVO> result = pagedUserIds.stream()
                    .map(userInfoMap::get)
                    .filter(Objects::nonNull)
                    .collect(Collectors.toList());
            
            log.info("获取点赞用户列表成功: resourceId={}, resourceType={}, 返回用户数={}, 总点赞数={}", 
                    resourceId, resourceType, result.size(), userIds.size());
            
            return PAIResponseBeanUtil.success(result);
            
        } catch (Exception e) {
            log.error("获取资源点赞用户列表异常: resourceId={}, resourceType={}", resourceId, resourceType, e);
            return PAIResponseBeanUtil.error(ResponseCodeEnum.ERROR.getCode(),"获取点赞用户列表失败");
        }
    }
    
    /**
     * 填充资源详细信息（复用AbstractResourceBehaviorService的逻辑）
     * 用于评论和回复通知中补充被评论资源的详细信息
     * 【关键修复】同时设置resourceImage和image字段，确保数据完整性
     *
     * @param messageNotifyVO 消息通知VO
     * @param resourceType    资源类型
     * @param resourceId      资源ID
     */
    private void fillResourceDetails(MessageNotifyVO messageNotifyVO, String resourceType, Long resourceId) {
        try {
            switch (resourceType) {
                case "MOMENT":
                    PAIResponseBean<MomentVO> momentResponse = momentClient.detail(resourceId);
                    if (PAIResponseBeanUtil.isOk(momentResponse) && momentResponse.getData() != null) {
                        MomentVO moment = momentResponse.getData();
                        // 设置动态内容摘要作为标题（截取前50个字符）
                        if (StringUtils.isNotBlank(moment.getContent())) {
                            String title = moment.getContent();
                            if (title.length() > 50) {
                                title = title.substring(0, 47) + "...";
                            }
                            messageNotifyVO.setResourceTitle(title);
                        }
                        // 设置封面图（取第一张图片）
                        if (!CollectionUtils.isEmpty(moment.getImages())) {
                            String imageUrl = String.valueOf(moment.getImages().get(0));
                            messageNotifyVO.setResourceImage(imageUrl);
                            // 【关键修复】同时设置image字段，用于消息列表展示
                            messageNotifyVO.setImage(imageUrl);
                        }
                        // 设置跳转链接
                        messageNotifyVO.setResourceUrl("/moment/detail/" + resourceId);
                    }
                    break;
                    
                case "ARTICLE":
                    PAIResponseBean<ArticleBaseResponseVO> articleResponse = subjectArticleClient.detail(resourceId, false);
                    if (PAIResponseBeanUtil.isOk(articleResponse) && articleResponse.getData() != null) {
                        ArticleBaseResponseVO article = articleResponse.getData();
                        // 设置文章标题
                        messageNotifyVO.setResourceTitle(article.getTitle());
                        // 设置文章封面图（防止cover为null导致空指针）
                        if (article.getCover() != null) {
                            String coverUrl = article.getCover().toString();
                            messageNotifyVO.setResourceImage(coverUrl);
                            // 【关键修复】同时设置image字段，用于消息列表展示
                            messageNotifyVO.setImage(coverUrl);
                        }
                        // 设置跳转链接
                        messageNotifyVO.setResourceUrl("/article/detail/" + resourceId);
                    }
                    break;
                    
                default:
                    // 其他资源类型可以在这里扩展
                    log.debug("未知的资源类型: {}", resourceType);
                    break;
            }
        } catch (Exception e) {
            log.warn("填充资源详细信息失败: resourceType={}, resourceId={}", resourceType, resourceId, e);
        }
    }
}
